package ltq.telphoneAgeAverage;

import ltq.util.DatabaseUtil;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

public class HotBrandBolt extends BaseRichBolt {
    private Map<String,Integer> hotBrandList;
    private int item,itemB;
    Connection conn=null;
    PreparedStatement ps = null;
    ResultSet rs = null;
    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
        hotBrandList=new HashMap<>();
        item=0;
        itemB=0;
        conn= DatabaseUtil.getConnection();
    }

    @Override
    public void execute(Tuple tuple) {
        String brand=tuple.getStringByField("brand");
        itemB=Integer.parseInt(tuple.getStringByField("flag"));
        if (brand.equals("null")){
            //空数据
            for (String key : hotBrandList.keySet()) {
//                refreshData(key,hotBrandList.get(key));
            }
            System.out.println(item);
            return;
        }
        if (hotBrandList.containsKey(brand)){
            int a=hotBrandList.get(brand)+1;
            hotBrandList.put(brand,a);
        }else {
            hotBrandList.put(brand,1);
        }
        item+=1;
        System.out.println("item2:"+item);
        if (item%itemB==0){
            for (String key : hotBrandList.keySet()) {
                refreshData(key,hotBrandList.get(key));
            }
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    private void refreshData(String brand, Integer count) {
        try {
            ps= conn.prepareStatement("select * from HotBrandResult where hBrand =?");
            ps.setObject(1,brand);
            rs= ps.executeQuery();
            if (rs.next()){
//                time2=rs.getLong("totalTime");
                ps= conn.prepareStatement("update HotBrandResult set hCount=? where hBrand=?");
                ps.setObject(1,count);
//                System.out.println("-----num------"+num);
                ps.setObject(2,brand);
                ps.executeUpdate();
            }
            else {
                ps= conn.prepareStatement("insert into HotBrandResult(hBrand,hCount) values (?,?)");
//                System.out.println("-----num------"+num);
                ps.setObject(1,brand);
                ps.setObject(2,count);
                ps.executeUpdate();
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }
}
